package org.code.rocketmq;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.code.common.Constant;
import org.code.util.ByteStringConverter;
import org.code.util.ThreadUtil;
import org.junit.Test;

import java.util.List;

/**
 * @author xiao
 * <p>
 * 发送单向消息
 * <p>
 * 消息发送方只负责发送消息，不等待服务端响应且没有回调函数触发，即只发送消息不等待响应。
 * 这种方式主要适用于某些耗时非常短，但对可靠性要求不是那么高的场景，例如日志收集。
 */
public class Single {
    /**
     * 主题
     */
    public static final String TOPIC = "single-topic";

    @Test
    public void singleProducer() throws Exception {
        // 创建默认消息生产者
        DefaultMQProducer producer = new DefaultMQProducer("single-producer-group");
        // 设置NameServer地址
        producer.setNamesrvAddr(Constant.NAME_SERVER);
        // 启动生产者
        producer.start();
        // 创建消息对象
        Message message = new Message(TOPIC, ByteStringConverter.stringToByteArray("single message"));
        // 发送消息
        // 由于在oneway方式发送消息时没有请求应答处理，如果出现消息发送失败，则会因为没有重试而导致数据丢失。若数据不可丢，建议选用可靠同步或可靠异步发送方式。
        producer.sendOneway(message);
        // 关闭生产者
        producer.shutdown();
    }

    @Test
    public void singleConsumer() throws Exception {
        // 创建默认消息消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("single-consumer-group");
        // 设置NameServer地址
        consumer.setNamesrvAddr(Constant.NAME_SERVER);
        // 设置订阅主题
        consumer.subscribe(TOPIC, "*");
        // 注册消息监听
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
                for (MessageExt messageExt : list) {
                    System.out.println(ThreadUtil.getCurrentThreadName() + " 接收到消息："
                            + ByteStringConverter.byteArrayToString(messageExt.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 启动消费者
        consumer.start();
        // 阻塞
        ThreadUtil.read();
    }
}
